package cn.com.sparkle.paxos;

import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.net.UnknownHostException;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import cn.com.sparkle.paxos.addprocess.AddRequestDealer;
import cn.com.sparkle.paxos.addprocess.speedcontrol.SpeedControlModel;
import cn.com.sparkle.paxos.admin.AdminLookupHandler;
import cn.com.sparkle.paxos.config.ConfigNode;
import cn.com.sparkle.paxos.config.Configuration;
import cn.com.sparkle.paxos.deamon.CatchUpDeamon;
import cn.com.sparkle.paxos.deamon.CheckMasterSenatorStateDeamon;
import cn.com.sparkle.paxos.deamon.HeartBeatCheckDeamon;
import cn.com.sparkle.paxos.deamon.InstanceExecutor;
import cn.com.sparkle.paxos.deamon.ReConnectDeamon;
import cn.com.sparkle.paxos.event.DefaultEventManager;
import cn.com.sparkle.paxos.event.EventsManager;
import cn.com.sparkle.paxos.handlerinterface.HandlerInterface;
import cn.com.sparkle.paxos.net.client.system.SystemClientHandler;
import cn.com.sparkle.paxos.net.netlayer.NetClient;
import cn.com.sparkle.paxos.net.netlayer.NetFactory;
import cn.com.sparkle.paxos.net.netlayer.NetServer;
import cn.com.sparkle.paxos.net.protocolprocessor.ProtocolManager;
import cn.com.sparkle.paxos.net.systemserver.SystemServerHandler;
import cn.com.sparkle.paxos.net.userserver.UserServerHandler;
import cn.com.sparkle.paxos.stablestorage.AccountBook;
import cn.com.sparkle.paxos.stablestorage.io.BufferedFileOut;
import cn.com.sparkle.paxos.state.ClusterState;
import cn.com.sparkle.paxos.state.NodeState;

public class ClassicPaxosServer {

	private final static Logger logger = Logger.getLogger(ClassicPaxosServer.class);
	private ReConnectDeamon reConnectThread = new ReConnectDeamon();
	private Long lastExceptInstanceId = null;
	private Context context;
	private DefaultEventManager eManager;

	public ClassicPaxosServer(Long lastExceptInstanceId) throws InstantiationException, IllegalAccessException {
		this.lastExceptInstanceId = lastExceptInstanceId;
		eManager = new DefaultEventManager();
		Runtime.getRuntime().addShutdownHook(new Thread() {
			@Override
			public void run() {
				logger.info("system exit~");
				LogManager.shutdown();// wait exit log
			}
		});
	}

	public ClassicPaxosServer() throws InstantiationException, IllegalAccessException {
		this(null);
	}

	public void init(String filePath, HandlerInterface userHandlerInterface) throws Throwable {
		Configuration configuration = new Configuration(filePath, eManager);
		ClusterState clusterState = new ClusterState(eManager, configuration);
		this.context = new Context(configuration, clusterState, eManager);

		// init BufferdFileOut
		BufferedFileOut.init(configuration.isDebugLog());

		ConfigNode configNode = getSelfConfig();
		SystemServerHandler handler = tryBind(configNode);
		InstanceExecutor ie = new InstanceExecutor(context, userHandlerInterface);
		// initiate account book
		AccountBook accountBook = new AccountBook(context, lastExceptInstanceId);
		context.setAccountBook(accountBook);
		// initiate instance executor thread
		ie.start();

		//init admin lookup processor
		AdminLookupHandler adminHandler = new AdminLookupHandler(context);

		//init protocol manager
		ProtocolManager protocolManager = ProtocolManager.createServerProtocol(context, userHandlerInterface, adminHandler);

		// initialize server handler
		handler.initProtocolManager(protocolManager);

		// initialize speed control
		new SpeedControlModel(context);
		// initiate heart beat deamon
		HeartBeatCheckDeamon beatCheckDeamon = new HeartBeatCheckDeamon(context);
		beatCheckDeamon.start();
		// initiate master check deamon
		Thread checkSenatorDeamon = new Thread(new CheckMasterSenatorStateDeamon(context));
		checkSenatorDeamon.setName("checkSenatorDeamon");
		checkSenatorDeamon.start();
		// initiate catch up deamon
		Thread catchUpDeamon = new Thread(new CatchUpDeamon(context));
		catchUpDeamon.setName("catchUpDeamon");
		catchUpDeamon.start();
		reConnectThread.startThread();
		// connect all node
		connectNodes(protocolManager);
		// init book
		accountBook.init(ie);
		// initiate user server
		userHandlerInterface.setAddRequestDealer(new AddRequestDealer(context));
		tryBindUserServer(userHandlerInterface, configNode);
	}

	public EventsManager getEventsManager() {
		return eManager;
	}

	private SystemClientHandler connectNodes(ProtocolManager protocolManager) throws Throwable {
		Configuration conf = context.getConfiguration();
		ClusterState cState = context.getcState();
		EventsManager eventsManager = context.getEventsManager();

		NetClient client = NetFactory.makeClient(conf.getNetLayer());
		SystemClientHandler phandler = new SystemClientHandler(client, eventsManager, reConnectThread, conf, protocolManager);
		client.init(conf.getFilePath() + "/system_out_net.prop", conf.getHeartBeatInterval(), phandler);

		NodesCollection senators = cState.getSenators();
		for (Entry<String, NodeState> s : senators.getNodeStates().entrySet()) {
			String[] temp = s.getKey().split(":");
			String ip = temp[0];
			String port = temp[1];
			client.connect(ip, Integer.parseInt(port), s.getKey());
		}
		return phandler;
	}

	private ConfigNode getSelfConfig() throws SocketException {
		HashSet<String> allIps = new HashSet<String>();
		String hostname = "";
		try {
			InetAddress ia = InetAddress.getLocalHost();
			hostname = ia.getHostName();
		} catch (UnknownHostException e) {
			logger.error("hostname can't be get!", e);
		}

		Enumeration<NetworkInterface> interfaceEnum = NetworkInterface.getNetworkInterfaces();
		while (interfaceEnum.hasMoreElements()) {
			NetworkInterface i = interfaceEnum.nextElement();
			Enumeration<InetAddress> ipEnum = i.getInetAddresses();
			while (ipEnum.hasMoreElements()) {
				allIps.add(ipEnum.nextElement().getHostAddress());
			}
		}
		Configuration conf = context.getConfiguration();
		Set<ConfigNode> set = conf.getConfigNodeSet().getSenators();

		for (ConfigNode s : set) {
			if ((allIps.contains(s.getIp()) || hostname.equals(s.getIp())) && String.valueOf(conf.getPort()).equals(s.getPort())) {
				conf.setSelfAddress(s.getAddress());
				return s;
			}
		}

		throw new RuntimeException("fail to find self address,check configuration's file please.");
	}

	private SystemServerHandler tryBind(ConfigNode configNode) throws Throwable {
		EventsManager eventsManager = context.getEventsManager();
		Configuration conf = context.getConfiguration();
		SystemServerHandler handler = new SystemServerHandler(eventsManager, conf);
		NetServer server = NetFactory.makeServer(conf.getNetLayer());
		server.init(conf.getFilePath() + "/system_in_net.prop", conf.getHeartBeatInterval(), handler);
		server.listen(configNode.getIp(), Integer.parseInt(configNode.getPort()));
		return handler;
	}

	private UserServerHandler tryBindUserServer(HandlerInterface userHandlerInterface, ConfigNode configNode) throws Throwable {
		EventsManager eventsManager = context.getEventsManager();
		Configuration conf = context.getConfiguration();
		UserServerHandler handler = new UserServerHandler(eventsManager, conf, userHandlerInterface, context.getProtocolManager());
		NetServer server = NetFactory.makeServer(conf.getNetLayer());
		server.init(conf.getFilePath() + "/service_in_net.prop", conf.getHeartBeatInterval(), handler);
		server.listen(configNode.getIp(), Integer.parseInt(configNode.getClientPort()));
		return handler;
	}
}
